package org.sunyaxing.transflow.transflowapp.plugins.inputs;

import cn.hutool.core.io.FileUtil;
import com.alibaba.fastjson2.JSONObject;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.sunyaxing.transflow.api.HandleData;
import org.sunyaxing.transflow.api.TransData;
import org.sunyaxing.transflow.api.common.Handle;
import org.sunyaxing.transflow.api.common.ano.Extension;
import org.sunyaxing.transflow.api.common.ano.HandleItem;
import org.sunyaxing.transflow.api.common.ano.JobParamItem;
import org.sunyaxing.transflow.api.common.ano.ScopeContentCheck;
import org.sunyaxing.transflow.api.extensions.base.ExtensionContext;
import org.sunyaxing.transflow.api.extensions.base.InputUtil;
import org.sunyaxing.transflow.api.extensions.base.types.TransFlowInput;
import org.sunyaxing.transflow.transflowapp.controllers.JobLogsWs;
import org.sunyaxing.transflow.transflowapp.plugins.inputs.nettyfile.FileStartInfo;
import org.sunyaxing.transflow.transflowapp.plugins.inputs.nettyfile.FileUploadInfo;
import org.sunyaxing.transflow.transflowapp.plugins.inputs.nettyfile.LastFileUploadInfo;

import java.io.File;
import java.io.RandomAccessFile;
import java.util.List;
import java.util.Map;
import java.util.function.Function;

@Extension("plugin-netty-file-input")
@ScopeContentCheck(value = {
        @JobParamItem(field = "host", label = "HOST", defaultValue = "0.0.0.0"),
        @JobParamItem(field = "port", label = "端口", defaultValue = "8080"),
        @JobParamItem(field = "tmp", label = "落盘路径", defaultValue = "/opt/transflow/fileTemp"),
}, handle = @HandleItem(field = "parentPath", label = "文件来源目录"))
public class NettyFileInputExt extends TransFlowInput<String, String> {
    private static final Logger log = LoggerFactory.getLogger(NettyFileInputExt.class);

    private EventLoopGroup boss;
    private EventLoopGroup worker;
    private ChannelFuture future;

    public NettyFileInputExt(ExtensionContext extensionContext) {
        super(extensionContext);
    }

    @Override
    public void commit(HandleData<String> handleData) {

    }

    @Override
    protected void afterInitHandler(JSONObject config, List<Handle> handles) {
        InputUtil<String> inputUtil = new InputUtil<>(this);
        String host = config.getString("host");
        Integer port = config.getInteger("port");
        String tmp = config.getString("tmp");
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        // 做是否支持epoll轮询判断以获取更高性能 EpollEventLoopGroup LINUX
        boss = Epoll.isAvailable() ? new EpollEventLoopGroup(1) : new NioEventLoopGroup(1);
        worker = Epoll.isAvailable() ? new EpollEventLoopGroup() : new NioEventLoopGroup();
        try {
            serverBootstrap.group(boss, worker)
                    .channel(Epoll.isAvailable() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childHandler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel ch) throws Exception {
                            ch.pipeline().addLast(new ObjectEncoder());
                            ch.pipeline().addLast(new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.weakCachingConcurrentResolver(null))); // 最大长度
                            ch.pipeline().addLast(new FileReceiverServerHandler(tmp, inputUtil));
                        }
                    });
            future = serverBootstrap.bind(host, port).sync();
            future.channel().closeFuture().addListener((ChannelFutureListener) channelFuture -> {
                channelFuture.channel().close();
            });
        } catch (Exception e) {
            log.error("e: ", e);
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        } finally {
        }
    }

    @Override
    public Function<TransData<String>, String> parseHandleToConsumer(String handleId, String path) {
        return stringTransData -> {
            JobLogsWs.sendMessage(jobId, "【{}】文件处理 {}", extensionContext.getNodeName(), stringTransData.getData());
            return stringTransData.getData();
        };
    }

    @Override
    public void destroy() {
        try {
            future.channel().close().sync();
        } catch (Exception e) {
            log.error("关闭netty错误", e);
        } finally {
            if (boss != null) {
                this.boss.shutdownGracefully();
            }
            if (worker != null) {
                this.worker.shutdownGracefully();
            }
        }
    }

    /**
     * 重写方法，通过目录匹配
     * @param handleValue
     * @return
     */
    @Override
    public String findHandleIdByValue(String handleValue) {
        for (Map.Entry<String, String> item : this.valueToHandleId.entrySet()) {
            if (item.getKey().contains(handleValue)) {
                return item.getValue();
            }
        }
        return null;
    }

    public static class FileReceiverServerHandler extends SimpleChannelInboundHandler<Object> {

        private String tmp;
        private FileStartInfo startInfo;
        private InputUtil<String> inputUtil;
        private RandomAccessFile randomAccessFile;

        public FileReceiverServerHandler(String tmp, InputUtil<String> inputUtil) {
            this.tmp = tmp;
            this.inputUtil = inputUtil;
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof FileStartInfo startInfo) {
                log.info("【接收端】准备接收文件 {}", startInfo.getFileName());
                this.startInfo = startInfo;
                File newFile = FileUtil.touch(this.tmp, startInfo.getFileName());
                this.randomAccessFile = new RandomAccessFile(newFile, "rw");
                this.randomAccessFile.seek(startInfo.getStartPos());
                // 回传已经准备好写入
                ctx.writeAndFlush(startInfo.getStartPos());
                TransData<String> transData = new TransData<>(0L, "接收到文件" + startInfo.getFileName());
                this.inputUtil.handle(this.startInfo.getFileParentPath(), transData);
            } else if (msg instanceof FileUploadInfo fileUploadInfo) {
                byte[] data = fileUploadInfo.getBytes();
                log.info("【接收端】从{}开始写入", fileUploadInfo.getStartPos());
                this.randomAccessFile.seek(fileUploadInfo.getStartPos());
                this.randomAccessFile.write(data, 0, fileUploadInfo.getByteSize());
                // 回传写入位置
                log.info("【接收端】当前进度 {}%", fileUploadInfo.getEndPos() * 100 / this.startInfo.getFileSize());
                ctx.writeAndFlush(fileUploadInfo.getEndPos());
            } else if (msg instanceof LastFileUploadInfo lastFileInfo) {
                log.info("【接收端】结束 {}", this.startInfo.getFileName());
                this.randomAccessFile.close();
                TransData<String> transData = new TransData<>(0L, "文件写入完毕" + startInfo.getFileName());
                this.inputUtil.handle(this.startInfo.getFileParentPath(), transData);
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            log.error("接收端异常", cause);
            try {
                ctx.close();
            } catch (Exception e) {
                log.error("关闭通道失败", e);
            }
            try {
                if (this.randomAccessFile != null) {
                    this.randomAccessFile.close();
                }
            } catch (Exception e) {
                log.error("关闭文件失败", e);
            }
        }
    }

}
